package com.swak.vertx.protocol.http.adapter;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.function.BiConsumer;
import java.util.function.Function;

import com.swak.utils.Lists;
import com.swak.utils.StringUtils;
import com.swak.vertx.cache.GlobalCacheService;
import com.swak.vertx.invoker.mark.CacheMark;
import com.swak.vertx.invoker.mark.HasCacheMark;
import com.swak.vertx.invoker.mark.MoreCacheMark;

/**
 * 参数的缓存支持
 * 
 * @author lifeng
 */
public class ParamCacheAdapter extends AbstractAdapter {

	private GlobalCacheService cacheStorage;
	private final Function<Context, CompletionStage<Context>> handle;

	public ParamCacheAdapter(GlobalCacheService cacheStorage) {
		this.cacheStorage = cacheStorage;
		this.handle = this::NextHandle;
	}

	@Override
	public CompletionStage<Context> handle(Context context) {
		CacheMark mark = context.$method.getParamCaches();
		CompletionStage<Context> future = null;
		if (mark instanceof HasCacheMark) {
			future = this.handleOne(context, (HasCacheMark) mark, (Object[]) context.$result);
		} else if (mark instanceof MoreCacheMark) {
			future = this.handleMore(context, (MoreCacheMark) mark, (Object[]) context.$result);
		}
		if (future != null) {
			return future.thenCompose(this.handle);
		}
		return this.next(context);
	}

	protected CompletionStage<Context> NextHandle(Context context) {
		return this.next(context);
	}

	protected CompletionStage<Context> handleOne(Context context, HasCacheMark mark, Object[] params) {
		String name = mark.paramValueForName() >= 0 ? String.valueOf(params[mark.paramValueForName()]) : mark.name();
		if (StringUtils.isBlank(name)) {
			return null;
		}
		CompletionStage<Object> future = this.cacheStorage.getObject(name);
		if (future != null) {
			return future.thenApply(res -> {
				params[mark.index()] = res;
				return context;
			}).exceptionally(e -> {
				context.$error = e;
				return context;
			});
		}
		return null;
	}

	protected CompletionStage<Context> handleMore(Context context, MoreCacheMark mark, Object[] params) {
		List<CacheMark> marks = mark.marks();
		List<CompletionStage<Context>> futures = Lists.newArrayList();
		for (CacheMark cMark : marks) {
			if (cMark instanceof HasCacheMark) {
				CompletionStage<Context> oneFutrue = this.handleOne(context, (HasCacheMark) cMark, params);
				if (oneFutrue != null) {
					futures.add(oneFutrue);
				}
			}
		}
		if (futures.size() == 1) {
			return futures.get(0);
		}
		return new ParallelFuture(context, futures).future();
	}

	/**
	 * 顺序任务
	 *
	 * @author lifeng
	 * @date 2020年9月16日 下午5:54:48
	 */
	private static class ParallelFuture {
		private List<CompletionStage<Context>> futures;
		private int index;
		private CompletableFuture<Context> future = new CompletableFuture<>();
		private final Context context;
		private final BiConsumer<Context, Throwable> handle;

		private ParallelFuture(Context context, List<CompletionStage<Context>> futures) {
			this.context = context;
			this.futures = futures;
			this.handle = this::handle;
		}

		/**
		 * 外部任务
		 *
		 * @return 异步结果
		 */
		private CompletionStage<Context> future() {

			// 执行内部任务
			this.innerFuture();

			// 返回异步结果
			return future;
		}

		/**
		 * 内部顺序执行任务
		 */
		private void innerFuture() {
			if (index >= futures.size()) {
				future.complete(this.context);
			}
			futures.get(index++).whenComplete(this.handle);
		}

		private void handle(Context context, Throwable error) {
			if (error != null) {
				future.completeExceptionally(error);
			} else {
				this.innerFuture();
			}
		}
	}
}
